package main

import (
	"context"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
	"minio.com/minio-sdk/erasure"
	"minio.com/minio-sdk/mask"
	"minio.com/minio-sdk/server"
	"minio.com/minio-sdk/storage"
	"minio.com/minio-sdk/util"
)

var (
	wg        sync.WaitGroup
	ORG_ID    = flag.String("i", "", "organization ID: 0, 1, 2, ...")
	ROLE      = flag.String("r", "help", "role of organization: server/client")
	WITHMASKS = flag.Bool("m", true, "provide data chunks with masks")
	PORT      = flag.String("p", "9090", "port of http server")
)

func do_server() {
	ctx := context.Background()

	// Initialize minio client object.
	minioClient, err := minio.New(util.Endpoint, &minio.Options{
		Creds:  credentials.NewStaticV4(util.AccessKeyID, util.SecretAccessKey, ""),
		Secure: util.UseSSL,
	})
	if err != nil {
		util.Logger.Fatalln(err)
	}

	// create minio bucket
	err = minioClient.MakeBucket(ctx, util.BUCKET_NAME, minio.MakeBucketOptions{Region: util.BUCKET_LOCATION})
	if err != nil {
		// Check to see if we already own this bucket (which happens if you run this twice)
		exists, errBucketExists := minioClient.BucketExists(ctx, util.BUCKET_NAME)
		if errBucketExists == nil && exists {
			util.Logger.Printf("We already own %s\n", util.BUCKET_NAME)
		} else {
			util.Logger.Fatalln(err)
		}
	} else {
		util.Logger.Printf("Successfully created %s\n", util.BUCKET_NAME)
	}

	// test if have data
	_, err = minioClient.StatObject(context.Background(), util.BUCKET_NAME, util.OBJ_NAME_PREFIX+"0", minio.StatObjectOptions{})
	util.Logger.Print("test if have data")
	if err != nil {
		util.Logger.Println("begin perpare data of minio:", err)
		storage.PerpareData(minioClient)
	}

	// load mask data
	util.Logger.Println("begin load mask data")
	maskData, err := mask.LoadMask(*ORG_ID)
	if err != nil {
		util.Logger.Println("load mask err:", err)
		return
	}
	util.Logger.Println("end load mask data, with mask:", *WITHMASKS)

	minioConfig := util.Minio_config{
		Endpoint:        util.Endpoint,
		AccessKeyID:     util.AccessKeyID,
		SecretAccessKey: util.SecretAccessKey,
		UseSSL:          util.UseSSL,
		Bucket:          util.BUCKET_NAME,
		WithMasks:       *WITHMASKS,
		MaskData:        maskData,
	}
	err = server.StartServer(minioConfig)
	if err != nil {
		util.Logger.Println("start server err:", err)
	}
}

func do_client() {
	ctx := context.Background()

	// Initialize minio client object.
	minioClient, err := minio.New(util.Endpoint, &minio.Options{
		Creds:  credentials.NewStaticV4(util.AccessKeyID, util.SecretAccessKey, ""),
		Secure: util.UseSSL,
	})
	if err != nil {
		util.Logger.Fatalln(err)
	}

	// init http client
	tr := &http.Transport{}
	cli := &http.Client{Transport: tr}

	k := util.K
	m := util.M
	masked_chunks := make([][]byte, k+m)
	parity_chunk_id := "3"
	wg.Add(k)
	time1 := time.Now().UnixNano()
	for i := 0; i < k; i++ {
		go func(cli *http.Client, data_chunk *[]byte, i int) {
			defer wg.Done()
			// util.Logger.Println("begin request data chunks from org:", i)
			oid := util.OBJ_NAME_PREFIX + strconv.Itoa(i) + "_obj_0"
			url := util.GetURL(i, util.BUCKET_NAME, oid)
			parity_chunk_id += "#" + oid
			resp, err := cli.Get(url)
			if err != nil {
				util.Logger.Printf("get url: %s err: %v\n", url, err)
			}
			defer resp.Body.Close()
			*data_chunk, err = ioutil.ReadAll(resp.Body)
			if err != nil {
				util.Logger.Println("read data err:", err)
			}
			// util.Logger.Printf("read data chunk from [org%d]\n", i)
		}(cli, &masked_chunks[i], i)
	}
	wg.Wait()
	time2 := time.Now().UnixNano()

	erasure.ErasureEncode(util.K, util.M, masked_chunks)
	time3 := time.Now().UnixNano()

	_, err = minioClient.PutObject(ctx, util.BUCKET_NAME, parity_chunk_id, strings.NewReader(string(masked_chunks[k+m-1])), util.OBJ_SIZE, minio.PutObjectOptions{})
	if err != nil {
		util.Logger.Println("write parity chunk to minio error:", err)
	}
	durationTime1 := (float64)(time2-time1) / 1e9
	durationTime2 := (float64)(time3-time2) / 1e9
	if *WITHMASKS {
		fmt.Printf("Read masked chunks from other orgs: %fs, Throughput: %fMB/s\n", durationTime1, (float64)(util.OBJ_SIZE)/(1024*1024*durationTime1))
		fmt.Printf("Encoding with masked data: %fs, Throughput: %fMB/s\n", durationTime2, (float64)(util.OBJ_SIZE)/(1024*1024*durationTime2))
	} else {
		fmt.Printf("Read original chunks from other orgs: %fs, Throughput: %fMB/s\n", durationTime1, (float64)(util.OBJ_SIZE)/(1024*1024*durationTime1))
		fmt.Printf("Encoding without masked data: %fs, Throughput: %fMB/s\n", durationTime2, (float64)(util.OBJ_SIZE)/(1024*1024*durationTime2))
	}

}

func main() {
	flag.Parse()

	// set log
	logFile, err := os.OpenFile(util.LoggerFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
	if err != nil {
		log.Println("open log file failed, err:", err)
		return
	}
	defer logFile.Close()
	util.Logger = log.New(logFile, "[main] -> ", log.Lshortfile|log.Ldate|log.Ltime)

	if *ORG_ID == "" {
		util.Logger.Println("Need org ID")
		return
	}
	switch *ROLE {
	case "server": // provide data chunks with/without masks
		util.OBJ_NAME_PREFIX += *ORG_ID + "_obj_"
		util.SERVER_PORT = ":" + *PORT
		do_server()
	case "client": // request data chunks and calculate parity chunks
		do_client()
	default:
		util.Logger.Println("Invalid role of organization")
		return
	}
}
